Skip to main content
Version: Next

Using the LangChain Transformer

LangChain is a software development framework designed to simplify the creation of applications using large language models (LLMs). Chains in LangChain go beyond just a single LLM call and are sequences of calls (can be a call to an LLM or a different utility), automating the execution of a series of calls and actions. To make it easier to scale up the LangChain execution on a large dataset, we have integrated LangChain with the distributed machine learning library SynapseML. This integration makes it easy to use the Apache Spark distributed computing framework to process millions of data with the LangChain Framework.

This tutorial shows how to apply LangChain at scale for paper summarization and organization. We start with a table of arxiv links and apply the LangChain Transformerto automatically extract the corresponding paper title, authors, summary, and some related works.

Step 1: Prerequisites​

The key prerequisites for this quickstart include a working Azure OpenAI resource, and an Apache Spark cluster with SynapseML installed. We suggest creating a Synapse workspace, but an Azure Databricks, HDInsight, or Spark on Kubernetes, or even a python environment with the pyspark package will work.

  1. An Azure OpenAI resource – request access here before creating a resource
  2. Create a Synapse workspace
  3. Create a serverless Apache Spark pool

Step 2: Import this guide as a notebook​

The next step is to add this code into your Spark cluster. You can either create a notebook in your Spark platform and copy the code into this notebook to run the demo. Or download the notebook and import it into Synapse Analytics

  1. Import the notebook into Microsoft Fabric, Synapse Workspace or if using Databricks into the Databricks Workspace.
  2. Install SynapseML on your cluster. Please see the installation instructions for Synapse at the bottom of the SynapseML website. Note that this requires pasting an additional cell at the top of the notebook you just imported.
  3. Connect your notebook to a cluster and follow along, editing and running the cells below.
%pip install -U openai==2.47.0 langchain-openai==1.4.0 langchain-community==0.4.2 pdf2image pdfminer.six unstructured==0.10.24 pytesseract nltk==3.8.1
from langchain_community.document_loaders import OnlinePDFLoader
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI
from synapse.ml.core.platform import find_secret

Step 3: Fill in the service information and construct the LLM​

Next, edit the cell to point to your Azure OpenAI v1 endpoint and deployment. You can replace find_secret with your key as follows:

openai_api_key = "99sj2w82o...."

openai_api_key = find_secret(
    secret_name="openai-api-key-3", keyvault="mmlspark-build-keys"
)
openai_base_url = "https://synapseml-openai-3.openai.azure.com/openai/v1/"
deployment_name = "gpt-5-mini"

llm = ChatOpenAI(
    model=deployment_name,
    base_url=openai_base_url,
    api_key=openai_api_key,
    max_completion_tokens=1024,
    reasoning_effort="low",
)

Step 4: Basic Usage of LangChain​

Create a chain​

We will start by demonstrating the basic usage with a simple chain that creates definitions for input words

copy_prompt = PromptTemplate(
    input_variables=["technology"],
    template="Define the following word: {technology}",
)

chain = {"technology": RunnablePassthrough()} | copy_prompt | llm | StrOutputParser()

technologies = ["docker", "spark", "python"]
definitions = chain.batch(technologies)

Create a Spark DataFrame from the results​

df = spark.createDataFrame(
    [
        (index, technology, str(definition))
        for index, (technology, definition) in enumerate(zip(technologies, definitions))
    ],
    ["label", "technology", "definition"],
)
display(df)

Serialization note​

Modern ChatOpenAI clients own HTTP connection pools and are not Spark-picklable. LangchainTransformer now rejects captured OpenAI clients instead of silently changing authentication or transport settings. Use LangChain batching as shown here, or use SynapseML's native OpenAI transformers for distributed inference.

Step 5: Using LangChain for literature review​

Create a Runnable pipeline for paper summarization​

We will construct a Runnable pipeline that loads an arXiv PDF and extracts its title, authors, and a short summary.

The pipeline contains these steps:

  1. OnlinePDFLoader: Load the first two PDF pages.
  2. PromptTemplate: Request the title, authors, and summary.
  3. ChatOpenAI: Generate the structured paper description.
  4. StrOutputParser: Return plain text for a Spark DataFrame.
def paper_content_extraction(arxiv_link: str) -> str:
    loader = OnlinePDFLoader(arxiv_link)
    pages = loader.load_and_split()
    content = "\n".join(page.page_content for page in pages[:2])
    return content


paper_summarizer_template = """Extract the paper title, authors, and a concise summary.
Here is the paper content:
{paper_content}
"""
paper_prompt = PromptTemplate.from_template(paper_summarizer_template)
paper_summary_chain = (
    {"paper_content": RunnablePassthrough()} | paper_prompt | llm | StrOutputParser()
)

Run the literature-review chain in a batch​

Use LangChain batching on the driver, then materialize the results as a Spark DataFrame.

papers = [
    (0, "https://arxiv.org/pdf/2107.13586.pdf"),
    (1, "https://arxiv.org/pdf/2101.00190.pdf"),
    (2, "https://arxiv.org/pdf/2103.10385.pdf"),
    (3, "https://arxiv.org/pdf/2110.07602.pdf"),
]
paper_contents = [paper_content_extraction(link) for _, link in papers]
paper_summaries = paper_summary_chain.batch(paper_contents)
paper_df = spark.createDataFrame(
    [
        (label, link, str(summary))
        for (label, link), summary in zip(papers, paper_summaries)
    ],
    ["label", "arxiv_link", "paper_info"],
)
display(paper_df)